package cc.eddic.practice.concurrency.impl;

import cc.eddic.practice.concurrency.ExecutorServiceUtil;
import cc.eddic.practice.concurrency.LogItem;
import cc.eddic.practice.concurrency.LogItemParser;
import cc.eddic.practice.concurrency.LogParser;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

@Slf4j
@Component
@Qualifier("LogParserWithCountDownLatchImpl")
public class LogParserWithCountDownLatchImpl implements LogParser {

    @Override
    public Map<String, List<LogItem>> parse(String folderName) {
        Map<String, List<LogItem>> result = new ConcurrentHashMap<>();

        ExecutorService executorService = Executors.newFixedThreadPool(10);
        CountDownLatch latch;
        try {
            val p = Paths.get(folderName);
            try (val walk = Files.walk(p)) {
                val list = walk.filter(Files::isRegularFile)
//                        .filter(x -> x.getFileName().toString().startsWith("2023"))
                        .filter(x-> x.getFileName().toString().endsWith(".log"))
                        .collect(Collectors.toList());
                latch = new CountDownLatch(list.size());
                CountDownLatch finalLatch = latch;

                list.forEach(path -> executorService.submit(() -> parse(path, finalLatch, result)));
            }
        } catch (IOException e) {
            log.error("File does not exists : {}", e.getMessage());
            return result;
        }

        try {
            if (log.isInfoEnabled()) log.info("await latch");
            latch.await();
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
        ExecutorServiceUtil.shutdown(executorService);
        return result;
    }

    private List<LogItem> parse(Path path, CountDownLatch cdl, Map<String, List<LogItem>> result ) {
        try (val lines = Files.lines(path)) {
            val items = lines.map(logItemParser::parse)
                    .filter(Optional::isPresent)
                    .map(Optional::get)
                    .collect(Collectors.toList());
            if (log.isInfoEnabled()) log.info("{} => {} rows", path, items.size());
            cdl.countDown();

            result.put(path.toString(), items);
            return items;
        } catch (IOException ex) {
            log.error(ex.getMessage(), ex);
            return new ArrayList<>();
        }
    }

    public LogParserWithCountDownLatchImpl(LogItemParser logItemParser) {
        this.logItemParser = logItemParser;
    }

    private final LogItemParser logItemParser;
}
